package flow

import (
	"context"
	"errors"
	"flow-calc/common"
	"flow-calc/config"
	"flow-calc/core/log"
	"fmt"
	"time"
)

/**
 *    Description: Flow流数据流
 *    Date: 2024/3/28
 */

// CommitRow 提交flow数据
func (flow *Flow) CommitRow(row interface{}) error {
	flow.buffer = append(flow.buffer, row)
	return nil
}

// commitSrcData 提交当前flow数据源数据,表示首次提交当前flow原始数据源
// 将flow的buffer中的数据提交到flow的data中
func (flow *Flow) commitSrcData(ctx context.Context) error {
	dataCnt := len(flow.buffer)
	batch := make(common.DataRowArray, 0, dataCnt)
	for _, dataRow := range flow.buffer {
		batch = append(batch, dataRow)
	}
	// 清空之前的数据源
	flow.clearData(flow.data)
	flow.data[common.FunctionIdFirstVirtual] = batch
	// 清空buffer
	flow.buffer = flow.buffer[0:0] // 该写法和nil是一样的效果，但是会复用之前的底层数组指针，减少内存分配等操作
	log.Logger().DebugCtx(ctx, "====> After commitSrcData, flow_name = %s, flow_id = %s\n 初始化首次源数据参数 = %+v\n", flow.Name, flow.Id, flow.data)
	return nil
}

// clearData 清空flow所有数据源
func (flow *Flow) clearData(data common.DataRowMap) {
	for key := range data {
		delete(data, key)
	}
}

// commitCurData 提交flow当前执行的Function 结果数据
func (flow *Flow) commitCurData(ctx context.Context) error {
	if len(flow.buffer) == 0 {
		flow.abort = true // 本次计算结果没有数据,设置中断退出flow流
		return nil
	}
	// 制作批量数据batch
	batch := make(common.DataRowArray, 0, len(flow.buffer))
	for _, dataRow := range flow.buffer {
		batch = append(batch, dataRow)
	}
	// 添加参数数据到函数中
	flow.data[flow.ExecFunctionId] = batch
	//清空缓冲Buf
	flow.buffer = flow.buffer[0:0]
	log.Logger().DebugCtx(ctx, " ====> After commitCurData, flow_name = %s, flow_id = %s\nAll Level Data =\n %+v\n", flow.Name, flow.Id, flow.data)
	return nil
}

// commitReuseData 提交flow上层结果数据,进行复用数据到下层
func (flow *Flow) commitReuseData(ctx context.Context) error {
	// 判断上层是否有结果数据
	if len(flow.data[flow.PrevFunctionId]) == 0 {
		flow.abort = true
		return nil
	}
	// 本层结果数据直接等于上层的结果数据。（达到复用结果数据目的）
	flow.data[flow.ExecFunctionId] = flow.data[flow.PrevFunctionId]
	// 清空buffer
	flow.buffer = flow.buffer[0:0]
	log.Logger().DebugCtx(ctx, " ====> After commitReuseData, flow_name = %s, flow_id = %s\nAll Level Data =\n %+v\n", flow.Name, flow.Id, flow.data)
	return nil
}

// commitVoidData 提交空数据
func (flow *Flow) commitVoidData(ctx context.Context) error {
	if len(flow.buffer) != 0 {
		return nil
	}
	// 制作空数据
	batch := make(common.DataRowArray, 0)
	// 将本层计算的缓冲数据提交到本层结果数据中
	flow.data[flow.ExecFunctionId] = batch
	log.Logger().DebugCtx(ctx, " ====> After commitVoidData, flow_name = %s, flow_id = %s\nAll Level Data =\n %+v\n", flow.Name, flow.Id, flow.data)
	return nil
}

// getCurData 获取当前Function 函数获取参数数据(采用的flow流式计算,这里获取的是上一层Function的输出结果数据)
func (flow *Flow) getCurData() (common.DataRowArray, error) {
	if flow.PrevFunctionId == "" {
		return nil, errors.New("flow.PrevFunctionId is nil")
	}
	if _, ok := flow.data[flow.PrevFunctionId]; !ok {
		return nil, errors.New(fmt.Sprintf("[%s] is not in flow.data", flow.PrevFunctionId))
	}
	return flow.data[flow.PrevFunctionId], nil
}

func (flow *Flow) Input() common.DataRowArray {
	return flow.inPut
}

// GetCacheData 获取缓存数据
func (flow *Flow) GetCacheData(key string) interface{} {
	if data, ok := flow.cache.Get(key); ok {
		return data
	}
	return nil
}

// SetCacheData 设置缓存数据
func (flow *Flow) SetCacheData(key string, value interface{}, expire time.Duration) {
	if expire == common.DeFaultExpiration {
		flow.cache.SetDefault(key, value)
	} else {
		flow.cache.Set(key, value, expire)
	}
}

// GetMetaData 获取元数据
func (flow *Flow) GetMetaData(key string) interface{} {
	flow.mLock.RLock()
	defer flow.mLock.RUnlock()
	data, ok := flow.metaData[key]
	if ok {
		return data
	}
	return nil
}

// SetMetaData 设置元数据对象的临时数据
func (flow *Flow) SetMetaData(key string, value interface{}) {
	flow.mLock.Lock()
	defer flow.mLock.Unlock()
	flow.metaData[key] = value
}

// GetFuncParam 得到Flow的当前执行Function的配置默认参数
func (flow *Flow) GetFuncParam(key string) string {
	flow.fpLock.RLock()
	defer flow.fpLock.RUnlock()
	if param, ok := flow.funcParams[flow.ExecFunctionId]; ok {
		if value, isExist := param[key]; isExist {
			return value
		}
	}
	return ""
}

// GetFuncParamAll 得到Flow的当前执行Function的配置全部默认参数
func (flow *Flow) GetFuncParamAll() config.FParam {
	flow.fpLock.RLock()
	defer flow.fpLock.RUnlock()
	param, ok := flow.funcParams[flow.ExecFunctionId]
	if !ok {
		return nil
	}
	return param
}

// GetFuncParamsAllFuncs 得到Flow的当前执行Function的配置全部默认参数
func (flow *Flow) GetFuncParamsAllFuncs() map[string]config.FParam {
	flow.fpLock.RLock()
	defer flow.fpLock.RUnlock()
	return flow.funcParams
}
